package com.ihr360.job.core.job;

import com.ihr360.job.core.BatchStatus;
import com.ihr360.job.core.ExitStatus;
import com.ihr360.job.core.Job;
import com.ihr360.job.core.JobExecutionException;
import com.ihr360.job.core.JobExecutionListener;
import com.ihr360.job.core.JobInterruptedException;
import com.ihr360.job.core.JobParametersIncrementer;
import com.ihr360.job.core.JobParametersValidator;
import com.ihr360.job.core.StartLimitExceededException;
import com.ihr360.job.core.Step;
import com.ihr360.job.core.entity.JobExecution;
import com.ihr360.job.core.entity.StepExecution;
import com.ihr360.job.core.job.builder.JobBuilderHelper;
import com.ihr360.job.core.lanuch.ExitCodeMapper;
import com.ihr360.job.core.lanuch.NoSuchJobException;
import com.ihr360.job.core.lanuch.support.JobRestartException;
import com.ihr360.job.core.listener.CompositeJobExecutionListener;
import com.ihr360.job.core.repeat.RepeatException;
import com.ihr360.job.core.repository.JobRepository;
import com.ihr360.job.core.scope.JobSynchronizationManager;
import com.ihr360.job.core.step.StepLocator;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.util.Assert;
import org.springframework.util.ClassUtils;

import java.util.Collection;
import java.util.Date;

public abstract class AbstractJob implements Job, StepLocator, BeanNameAware,
        InitializingBean {
    protected final Logger logger = LoggerFactory.getLogger(JobBuilderHelper.class.getName());
    private String name;

    private boolean restartable = true;

    private JobRepository jobRepository;

    private CompositeJobExecutionListener listener = new CompositeJobExecutionListener();

    private JobParametersIncrementer jobParametersIncrementer;

    private JobParametersValidator jobParametersValidator = new DefaultJobParametersValidator();

    private StepHandler stepHandler;

    /**
     * Default constructor.
     */
    public AbstractJob() {
        super();
    }
    /**
     * Convenience constructor to immediately add name (which is mandatory but
     * not final).
     *
     * @param name
     */
    public AbstractJob(String name) {
        super();
        this.name = name;
    }

    /**
     * A validator for job parameters. Defaults to a vanilla
     * {@link DefaultJobParametersValidator}.
     *
     * @param jobParametersValidator
     *            a validator instance
     */
    public void setJobParametersValidator(
            JobParametersValidator jobParametersValidator) {
        this.jobParametersValidator = jobParametersValidator;
    }

    /**
     * Assert mandatory properties: {@link JobRepository}.
     *
     * @see InitializingBean#afterPropertiesSet()
     */
    @Override
    public void afterPropertiesSet() throws Exception {
        Assert.notNull(jobRepository, "JobRepository must be set");
    }

    /**
     * Set the name property if it is not already set. Because of the order of
     * the callbacks in a Spring container the name property will be set first
     * if it is present. Care is needed with bean definition inheritance - if a
     * parent bean has a name, then its children need an explicit name as well,
     * otherwise they will not be unique.
     *
     * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String)
     */
    @Override
    public void setBeanName(String name) {
        if (this.name == null) {
            this.name = name;
        }
    }

    /**
     * Set the name property. Always overrides the default value if this object
     * is a Spring bean.
     *
     * @see #setBeanName(java.lang.String)
     */
    public void setName(String name) {
        this.name = name;
    }

    /*
     * (non-Javadoc)
     *
     * @see org.springframework.batch.core.domain.IJob#getName()
     */
    @Override
    public String getName() {
        return name;
    }

    /**
     * Retrieve the step with the given name. If there is no Step with the given
     * name, then return null.
     *
     * @param stepName
     * @return the Step
     */
    @Override
    public abstract Step getStep(String stepName);

    /**
     * Retrieve the step names.
     *
     * @return the step names
     */
    @Override
    public abstract Collection<String> getStepNames();

    @Override
    public JobParametersValidator getJobParametersValidator() {
        return jobParametersValidator;
    }

    /**
     * Boolean flag to prevent categorically a job from restarting, even if it
     * has failed previously.
     *
     * @param restartable
     *            the value of the flag to set (default true)
     */
    public void setRestartable(boolean restartable) {
        this.restartable = restartable;
    }

    /**
     * @see Job#isRestartable()
     */
    @Override
    public boolean isRestartable() {
        return restartable;
    }

    /**
     * Public setter for the {@link JobParametersIncrementer}.
     *
     * @param jobParametersIncrementer
     *            the {@link JobParametersIncrementer} to set
     */
    public void setJobParametersIncrementer(
            JobParametersIncrementer jobParametersIncrementer) {
        this.jobParametersIncrementer = jobParametersIncrementer;
    }

    /*
     * (non-Javadoc)
     *
     * @see org.springframework.batch.core.Job#getJobParametersIncrementer()
     */
    @Override
    public JobParametersIncrementer getJobParametersIncrementer() {
        return this.jobParametersIncrementer;
    }

    /**
     * Public setter for injecting {@link JobExecutionListener}s. They will all
     * be given the listener callbacks at the appropriate point in the job.
     *
     * @param listeners
     *            the listeners to set.
     */
    public void setJobExecutionListeners(JobExecutionListener[] listeners) {
        for (int i = 0; i < listeners.length; i++) {
            this.listener.register(listeners[i]);
        }
    }

    /**
     * Register a single listener for the {@link JobExecutionListener}
     * callbacks.
     *
     * @param listener
     *            a {@link JobExecutionListener}
     */
    public void registerJobExecutionListener(JobExecutionListener listener) {
        this.listener.register(listener);
    }

    /**
     * Public setter for the {@link JobRepository} that is needed to manage the
     * state of the batch meta domain (jobs, steps, executions) during the life
     * of a job.
     *
     * @param jobRepository
     */
    public void setJobRepository(JobRepository jobRepository) {
        this.jobRepository = jobRepository;
        stepHandler = new SimpleStepHandler(jobRepository);
    }

    /**
     * Convenience method for subclasses to access the job repository.
     *
     * @return the jobRepository
     */
    protected JobRepository getJobRepository() {
        return jobRepository;
    }

    /**
     * Extension point for subclasses allowing them to concentrate on processing
     * logic and ignore listeners and repository calls. Implementations usually
     * are concerned with the ordering of steps, and delegate actual step
     * processing to {@link #handleStep(Step, JobExecution)}.
     *
     * @param execution
     *            the current {@link JobExecution}
     *
     * @throws JobExecutionException
     *             to signal a fatal batch framework error (not a business or
     *             validation exception)
     */
    abstract protected void doExecute(JobExecution execution)
            throws JobExecutionException;

    /**
     * Run the specified job, handling all listener and repository calls, and
     * delegating the actual processing to {@link #doExecute(JobExecution)}.
     *
     * @see Job#execute(JobExecution)
     * @throws StartLimitExceededException
     *             if start limit of one of the steps was exceeded
     */
    @Override
    public final void execute(JobExecution execution) {

        if (logger.isDebugEnabled()) {
            logger.debug("Job execution starting: " + execution);
        }

        JobSynchronizationManager.register(execution);

        try {

            jobParametersValidator.validate(execution.getJobParameters());

            if (execution.getStatus() != BatchStatus.STOPPING) {

                execution.setStartTime(new Date());
                updateStatus(execution, BatchStatus.STARTED);

                listener.beforeJob(execution);

                try {
                    doExecute(execution);
                    if (logger.isDebugEnabled()) {
                        logger.debug("Job execution complete: " + execution);
                    }
                } catch (RepeatException e) {
                    throw e.getCause();
                }
            } else {

                // The job was already stopped before we even got this far. Deal
                // with it in the same way as any other interruption.
                execution.setStatus(BatchStatus.STOPPED);
                execution.setExitStatus(ExitStatus.COMPLETED);
                if (logger.isDebugEnabled()) {
                    logger.debug("Job execution was stopped: " + execution);
                }

            }

        } catch (JobInterruptedException e) {
            logger.info("Encountered interruption executing job: "
                    + e.getMessage());
            if (logger.isDebugEnabled()) {
                logger.debug("Full exception", e);
            }
            execution.setExitStatus(getDefaultExitStatusForFailure(e, execution));
            execution.setStatus(BatchStatus.max(BatchStatus.STOPPED, e.getStatus()));
            execution.addFailureException(e);
        } catch (Throwable t) {
            logger.error("Encountered fatal error executing job", t);
            execution.setExitStatus(getDefaultExitStatusForFailure(t, execution));
            execution.setStatus(BatchStatus.FAILED);
            execution.addFailureException(t);
        } finally {
            try {
                if (execution.getStatus().isLessThanOrEqualTo(BatchStatus.STOPPED)
                        && execution.getStepExecutions().isEmpty()) {
                    ExitStatus exitStatus = execution.getExitStatus();
                    ExitStatus newExitStatus =
                            ExitStatus.NOOP.addExitDescription("All steps already completed or no steps configured for this job.","任务已完成或者没有为任务添加任何计算步骤");
                    execution.setExitStatus(exitStatus.and(newExitStatus));
                }

                execution.setEndTime(new Date());

                try {
                    listener.afterJob(execution);
                } catch (Exception e) {
                    logger.error("Exception encountered in afterStep callback", e);
                }

                jobRepository.update(execution);
            } finally {
                JobSynchronizationManager.release();
            }

        }

    }

    /**
     * Convenience method for subclasses to delegate the handling of a specific
     * step in the context of the current {@link JobExecution}. Clients of this
     * method do not need access to the {@link JobRepository}, nor do they need
     * to worry about populating the execution context on a restart, nor
     * detecting the interrupted state (in job or step execution).
     *
     * @param step
     *            the {@link Step} to execute
     * @param execution
     *            the current {@link JobExecution}
     * @return the {@link StepExecution} corresponding to this step
     *
     * @throws JobInterruptedException
     *             if the {@link JobExecution} has been interrupted, and in
     *             particular if {@link BatchStatus#ABANDONED} or
     *             {@link BatchStatus#STOPPING} is detected
     * @throws StartLimitExceededException
     *             if the start limit has been exceeded for this step
     * @throws JobRestartException
     *             if the job is in an inconsistent state from an earlier
     *             failure
     */
    protected final StepExecution handleStep(Step step, JobExecution execution)
            throws JobInterruptedException, JobRestartException,
            StartLimitExceededException {
        return stepHandler.handleStep(step, execution);

    }

    /**
     * Default mapping from throwable to {@link ExitStatus}.
     *
     * @param ex
     *            the cause of the failure
     * @return an {@link ExitStatus}
     */
    protected ExitStatus getDefaultExitStatusForFailure(Throwable ex, JobExecution execution) {
        ExitStatus exitStatus;
        if (ex instanceof JobInterruptedException
                || ex.getCause() instanceof JobInterruptedException) {
            exitStatus = ExitStatus.STOPPED
                    .addExitDescription(JobInterruptedException.class.getName(),ex.getMessage());
        } else if (ex instanceof NoSuchJobException
                || ex.getCause() instanceof NoSuchJobException) {
            exitStatus = new ExitStatus(ExitCodeMapper.NO_SUCH_JOB, ex
                    .getClass().getName());
        } else {
            exitStatus = ExitStatus.FAILED.addExitDescription(ex);
        }

        return exitStatus;
    }

    private void updateStatus(JobExecution jobExecution, BatchStatus status) {
        jobExecution.setStatus(status);
        jobRepository.update(jobExecution);
    }

    @Override
    public String toString() {
        return ClassUtils.getShortName(getClass()) + ": [name=" + name + "]";
    }
}